package com.zhengb.flink.study.log.analysis.sink;

import com.alibaba.fastjson.JSONObject;
import com.zhengb.flink.study.log.analysis.entity.LogMessageSink;
import com.zhengbo.flink.demo.base.mongo.MongoService;
import com.zhengbo.flink.demo.base.utils.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * Created by zhengbo on 2020/2/6.
 */
@Slf4j
public class LogAnaylysisSinkFunction extends RichSinkFunction<LogMessageSink> {

    private MongoService mongoService;

    private final static String SINK_COLLECTION = "";

    @Override
    public void open(Configuration parameters) throws Exception {

        log.info("LogAnaylysisSinkFunction open:{}", parameters);
        super.open(parameters);

        mongoService = SpringContextUtil.getBean(MongoService.class);

        log.info("LogAnaylysisSinkFunction open success...");
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    @Override
    public void invoke(LogMessageSink value, Context context) throws Exception {

        log.info("insert sink data:{}", value);
        mongoService.insert(JSONObject.toJSONString(value), SINK_COLLECTION);
    }
}
